package spring.cloud.tasks.tasks_manager.schedule;

import com.alibaba.fastjson2.JSON;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.*;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.RenewalNodePaths;
import spring.cloud.tasks.common.nodepaths.TaskInstanceNodePaths;
import spring.cloud.tasks.common.status.TaskInstanceInfoStatus;
import spring.cloud.tasks.common.status.TaskItemStatus;
import spring.cloud.tasks.common.status.TaskStageStatus;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.BatchTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.StateFullSimpleTaskTaskInstanceManager;
import spring.cloud.tasks.tasks_manager.taskInstanceManager.StateLessSimpleTaskTaskInstanceManager;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 任务实例状态管理器：任务实例信息维护。
 * 1ZK通过监听器，动态监听实例信息变更。
 * 2启动线程定期维护线程实例状态。比如线程超时。在这期间线程实例对象需要加锁。
 * 3线程实例下个状态翻转。
 */
@Slf4j
public class TaskInstanceStateManager {
    protected ScheduleManager scheduleManager;
    private final String nameSpace;
    protected Curator curator;
    public Lock lock = new ReentrantLock();

    public TaskInstanceStateManager(ScheduleManager scheduleManager, String nameSpace, Curator curator) {
        this.scheduleManager = scheduleManager;
        this.nameSpace = nameSpace;
        this.curator = curator;
    }

    private long currentTimeStamp = System.currentTimeMillis();

    /**
     * 入口
     */
    public void refresh() throws Exception {
        currentTimeStamp = System.currentTimeMillis();
        /**
         * 任务实例状态+任务实例stage状态变更，任务实例stage状态变更后，对之前阶段的任务分片进行清理，对之后的阶段的任务分片进行部署。
         * 这里需要考虑的是，有的任务分片，只要预分配了，执行器检测到该任务分片，则需要立即执行。比如捞取批次开始则需要按照指定的时间执行，以保证数据能够完全的准备好。
         * 后面的阶段则不用进行按照指定的时间进行准备了，有数据则行。
         */
        {
            BatchTaskTaskInstanceManager batchTaskInstanceManager = scheduleManager.batchTaskTaskInstanceManager;
            ConcurrentHashMap<String, CopyOnWriteArrayList<TaskInstanceInfo>> batchTaskRunningTask = batchTaskInstanceManager.assignedTasks;
            //
            refreshByTaskType(batchTaskRunningTask);
        }
        {
            StateFullSimpleTaskTaskInstanceManager simpleTaskInstanceManager = scheduleManager.stateFullSimpleTaskTaskInstanceManager;
            ConcurrentHashMap<String, CopyOnWriteArrayList<TaskInstanceInfo>> simpleTaskStateFullRunningTask = simpleTaskInstanceManager.assignedTasks;
            refreshByTaskType(simpleTaskStateFullRunningTask);
        }
        {
            StateLessSimpleTaskTaskInstanceManager simpleTaskStateLessInstanceManager = scheduleManager.stateLessSimpleTaskTaskInstanceManager;
            ConcurrentHashMap<String, CopyOnWriteArrayList<TaskInstanceInfo>> simpleTaskStateLessRunningTask = simpleTaskStateLessInstanceManager.assignedTasks;
            //
            refreshByTaskType(simpleTaskStateLessRunningTask);
        }
        //
        Set<String> taskIdSet = TaskTouchInfo.Context.getTaskIdSet();
        for (String taskId : taskIdSet) {
            TaskTouchInfo taskTouchInfo = TaskTouchInfo.Context.get(taskId);
            if (taskTouchInfo != null) {
                //触碰信息本身的遗漏清理，避免内存溢出。
                //为了保证触碰信息不错乱清理，那么需要关于任务以及任务实例触碰必须及时+正确
                if (currentTimeStamp - taskTouchInfo.lastTouchTimeStamp > DateUtils.DAY * 2) {
                    TaskTouchInfo.Context.remove(taskId);
                }
            } else {
                //这种情况不存在。因为只会在这个类进行remove add by huihui
                TaskTouchInfo.Context.remove(taskId);
            }
        }

    }

    private void refreshByTaskType(ConcurrentHashMap<String, CopyOnWriteArrayList<TaskInstanceInfo>> map) {
        Set<String> taskIdSet = map.keySet();
        for (String taskId : taskIdSet) {
            currentTimeStamp = System.currentTimeMillis();//每处理一个任务更新一次当前的时间戳
            List<TaskInstanceInfo> taskInstanceInfoList = map.get(taskId);
            if (taskInstanceInfoList != null) {
                refresh(taskInstanceInfoList);
                tryRemoveTaskInstanceInfoList(map, taskId, taskInstanceInfoList);
            }
        }
    }


    private void refresh(List<TaskInstanceInfo> taskInstanceInfoList) {
        //原本想使用list.iterator().remove();这样清理对象性能快一些，但是为了避免并发问题（在使用迭代器迭代集合的同时,使用原集合修改元素；如果迭代器发现自己和集合不一样，就会抛出 ConcurrentModificationException 异常。）
        //解决方案有三种:
        //1使用列表迭代器list.listIterator();it.add("我全都要");
        //2不使用迭代器遍历,使用普通for遍历 (最优，使用双list循环来解决。)
        //3使用toArray
        for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
            refresh(taskInstanceInfo);
        }
        List<TaskInstanceInfo> list = new ArrayList<>();
        for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
            if (taskInstanceInfo.cleared) {
                list.add(taskInstanceInfo);
            }
        }
        for (TaskInstanceInfo taskInstanceInfo : list) {
            taskInstanceInfoList.remove(taskInstanceInfo);
        }
    }

    private void refresh(TaskInstanceInfo taskInstanceInfo) {
        try {
            /**
             * 任务实例超时清理。但是在超时清理前需要进行状态处理。这个和其他的地方不一样。
             */
            boolean clear = clear(taskInstanceInfo);
            if (clear) {
                return;
            }
            stateUpdate(taskInstanceInfo);
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        }
    }

    private boolean clear(TaskInstanceInfo taskInstanceInfo) {
        String status = taskInstanceInfo.getStatus();
        if (TaskInstanceInfoStatus.isTerminalState(status)) {
            if (taskInstanceInfo.isReachClearTime()) {
                taskInstanceInfo.setCleared(true);
                return true;
            }
        }
        return false;
    }

    private void stateUpdate(@NonNull TaskInstanceInfo taskInstanceInfo) {
        try {
            Lock lock = taskInstanceInfo.lock;
            boolean tryLock = lock.tryLock(15, TimeUnit.SECONDS);
            if (tryLock) {
                try {
                    stateUpdate0(taskInstanceInfo);
                } finally {
                    lock.unlock();
                }
            }

        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        }

    }

    /**
     * 任务实例状态，任务实例当前阶段状态，任务实例当前阶段-分片状态，从上至下进行状态检查
     *
     * @param taskInstanceInfo
     * @throws Exception
     */
    private void stateUpdate0(TaskInstanceInfo taskInstanceInfo) throws Exception {
        if (taskInstanceInfo.isTryCancel()) {
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
            taskInstanceInfo.setCancelled(true);//如果调度框架没有取消这个任务，可能这个任务还会执行一会。
            String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
            curator.merge(taskInstanceStatusNodePath, taskInstanceInfo.getStatus());
            return;
        }
        if (taskInstanceInfo.isCancelled()) {
            //如果任务实例被取消了 那么则需要暂停任务分片信息等
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
            String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
            curator.merge(taskInstanceStatusNodePath, taskInstanceInfo.getStatus());
            return;
            //任务实例的阶段状态不进行修改。

        }
        String status = taskInstanceInfo.getStatus();
        if (status == null) {
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_value_is_null);
            log.warn("任务实例状态为空:" + JSON.toJSONString(taskInstanceInfo));
            return;//代表状态出现了异常。因为任何一个时间点，任务实例的状态都不可能为null
        }
        //只有当任务实例状态为注册成功，那么才进行内存+ZK状态同步。
        if (!taskInstanceInfo.isRegistered()) {
            return;
        }
        /**
         * 可能状态为取消+调度结束(比如用户取消+超时)
         */
        if (TaskInstanceInfoStatus.isTerminalState(status)) {
            //终态补数据
            try {
                String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
                String taskInstanceStatus = curator.getData(taskInstanceStatusNodePath);
                if (TaskInstanceInfoStatus.waiting_task_item_pre_assign_success.equals(taskInstanceStatus)) {
                    curator.merge(taskInstanceStatusNodePath, status);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
            return;
        }
        //下面是现ZK更新状态后再更新内存。因为如果先更新ZK，再更新内存。
        // 如果先更新内存再更新ZK。内存成功，ZK失败。如果调度器异常，那么整个系统可能会存在一些难以预料的问题出现。
        // 整个调度ZK的前提是ZK网络正常。如果ZK网络异常，即使当前实例信息不更新。。。，但是其他任务实例信息的分片信息+实例信息ZK同步也会失败。
        String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
        if (curator.checkExists(taskInstanceStatusNodePath)) {

        } else {

        }
        String taskInstanceStatus = curator.getData(taskInstanceStatusNodePath);
        if (taskInstanceStatus == null) {
            //任务状态还没有进行同步
        }
        if (taskInstanceStatus != null) {
            if (status.equals(taskInstanceStatus)) {
                return;
            }
        }
        String currentStage = taskInstanceInfo.getCurrentStage();
        String currentStageStatus = taskInstanceInfo.getCurrentStageStatus();
        if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskInstanceInfo.getTaskType())) {
            //简单分片不用改状态。只要实例单个分片成功+以及有状态的分片不进行乱跑，以及单个分片超时，以及总的实例是否超时即可。
            //
            if (TaskInstanceInfoStatus.waiting_task_item_pre_assign_success.equals(status)) {//只有这个状态才进行状态维护
                //检查下面所有的分片是否都已经有状态了
                SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo = JSON.parseObject(taskInstanceInfo.getTaskItemInfo(), SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.class);
                int taskItemCount = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemCount;
                String rootPath = RenewalNodePaths.getRoot() + Symbols.SLASH + taskInstanceInfo.getTaskId() + Symbols.SLASH + taskInstanceInfo.getTaskInstanceId();
                boolean allTaskItemsIsTerminalState = true;
                for (int i = 1; i <= taskItemCount; i++) {
                    String path = rootPath + Symbols.SLASH + TaskType.SimpleSlicingTask.stage_one + Symbols.SLASH + (YN.Y.equals(taskInstanceInfo.getStateFullTask()) ? TaskType.SimpleSlicingTask.stage_type_state_full : TaskType.SimpleSlicingTask.stage_type_state_less) + Symbols.SLASH + i;
                    String statusPath = path + Symbols.SLASH + "status";
                    //String markPath = path + Symbols.SLASH + "mark";
                    boolean isTerminalState = TaskItemStatus.isTerminalState(curator.getData(statusPath));
                    if (!isTerminalState) {
                        allTaskItemsIsTerminalState = false;
                        break;
                    }
                }
                if (allTaskItemsIsTerminalState) {
                    //ZK先同步
                    curator.merge(taskInstanceStatusNodePath, TaskInstanceInfoStatus.status_completed);
                    //任务实例状态标记为失败。
                    taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_completed);
                }
                //分片已经分配了，看看是否超时。 可能这存在问题
                if (taskInstanceInfo.isTimeout()) {
                    //ZK先同步
                    curator.merge(taskInstanceStatusNodePath, TaskInstanceInfoStatus.status_timeout);
                    //任务实例状态标记为超时。
                    taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_timeout);
                    return;
                }
                if (YN.Y.equals(taskInstanceInfo.getStateFullTask())) {
                    //如果是有状态的，那么我们需要新增一个有状态的任务管理，把当前的状态通知给有状态的所有的任务实例。
                }
            }

        } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskInstanceInfo.getTaskType())//
                ||//
                TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskInstanceInfo.getTaskType())//
        ) {
            boolean isTerminalState = TaskStageStatus.isTerminalState(currentStageStatus);
            if (isTerminalState) {
                if (TaskStageStatus.status_success.equals(currentStageStatus)) {
                    //看看是否还有下一个阶段。
                    List<TaskStageInfo> taskStageAndTaskItemTypeAndCountAndLoadInfoList = JSON.parseArray(taskInstanceInfo.getTaskItemInfo(), TaskStageInfo.class);
                    Collections.sort(taskStageAndTaskItemTypeAndCountAndLoadInfoList, new Comparator<TaskStageInfo>() {
                        @Override
                        public int compare(TaskStageInfo o1, TaskStageInfo o2) {
                            return o1.stageNum - o2.stageNum;//越小放前面
                        }
                    });
                    int size = taskStageAndTaskItemTypeAndCountAndLoadInfoList.size();
                    Map<String, Integer> stageToStageNumMap = new HashMap<>();
                    Map<String, String> stageNumToStageMap = new HashMap<>();
                    for (TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoadInfo : taskStageAndTaskItemTypeAndCountAndLoadInfoList) {
                        String stage = taskStageAndTaskItemTypeAndCountAndLoadInfo.stage;
                        int stageNum = taskStageAndTaskItemTypeAndCountAndLoadInfo.stageNum;
                        stageToStageNumMap.put(stage, stageNum);
                        stageNumToStageMap.put(String.valueOf(stageNum), stage);
                    }
                    int stageNum = stageToStageNumMap.get(currentStage);
                    if (size == stageNum) {
                        //ZK先同步
                        curator.merge(taskInstanceStatusNodePath, TaskInstanceInfoStatus.status_success);
                        //任务实例状态标记为成功。
                        taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_success);
                        return;
                    }
                    String nextStage = stageNumToStageMap.get(String.valueOf(stageNum + 1));
                    taskInstanceInfo.setCurrentStage(nextStage);
                    taskInstanceInfo.setCurrentStageStatus(TaskStageStatus.status_watting_init);//初始化
                    //ZK同步
                    return;
                }
                if (TaskStageStatus.status_fail.equals(currentStageStatus)) {
                    //ZK先同步
                    curator.merge(taskInstanceStatusNodePath, TaskInstanceInfoStatus.status_fail);
                    //任务实例状态标记为失败。
                    taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_fail);
                    return;
                }
                if (TaskStageStatus.status_timeout.equals(currentStageStatus)) {
                    //ZK先同步
                    curator.merge(taskInstanceStatusNodePath, TaskInstanceInfoStatus.status_timeout);
                    //任务实例状态标记为超时。
                    taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_timeout);
                    return;
                }
            }
        }


        //
    }

    /**
     * 如果1天了没有进行触碰+没有任何任务实例存在则将该任务实例列表清理掉，避免内存溢出。
     */
    private void tryRemoveTaskInstanceInfoList(ConcurrentHashMap<String, CopyOnWriteArrayList<TaskInstanceInfo>> map, String taskId, List<TaskInstanceInfo> taskInstanceInfoList) {
        if (taskInstanceInfoList.size() == 0) {
            TaskTouchInfo taskTouchInfo = TaskTouchInfo.Context.get(taskId);
            if (taskTouchInfo != null) {
                //两天都没有进行触碰了 可以安全删除了
                if (currentTimeStamp - taskTouchInfo.lastTouchTimeStamp > DateUtils.DAY * 1) {
                    map.remove(taskId);
                    TaskTouchInfo.Context.remove(taskId);
                }
            }

        } else {
            //只要实例信息存活，则需要保持触碰信息存在。fix a bug,这个bug就是该任务实例信息不主动触碰，导致触碰信息比实例信息先删除。
            TaskTouchInfo.Context.put(taskId);
        }
    }
}
